package org.chnxi.study.messagequeue.rockermq.consumer;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.MessageSelector;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * 针对用户设置的Property中的数值进行SQL语法匹配：MessageSelector.bySql("a > 5")
 * 语法：
 * <ul>
 *     <li>数值比较：>、>=、<、<=、between、=</li>
 *     <li>字符比较：=、<>、in</li>
 *     <li>is null 或者 is not null</li>
 *     <li>逻辑符号：and、or、not</li>
 * </ul>
 * 注意事项：
 * <ul>
 *     <li>数值，比如：123、3.1415</li>
 *     <li>字符，比如：'abc'必须使用单引号包裹</li>
 *     <li>NULL，特殊的常量</li>
 *     <li>布尔值，true或false</li>
 * </ul>
 */
public class SqlFilterConsumer {
    public static void main(String[] args) throws MQClientException {
        //创建消费者对象
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer();
        //设置NameServer地址
        consumer.setNamesrvAddr("192.167.50.227:9876;192.167.50.227:9877");
        //设置消费者所在的分组
        consumer.setConsumerGroup("consumerGroup");
        //同一个消费者组内，使用广播模式时，组内所有成员都将受到消息
        //consumer.setMessageModel(MessageModel.CLUSTERING);
        //设置监听的Topic和tag
        consumer.subscribe("topicName" , MessageSelector.bySql("a > 5"));
        //设置接受消息处理函数
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list,
                                                            ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for (MessageExt msgExt : list) {
                    System.out.println("Message："+new String(msgExt.getBody()));
                }
                //直接返回消费成功
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
    }
}
